package org.yonggan.flume;

import org.apache.commons.io.FileUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.charset.Charset;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 自动用source 源头
 */

public class TailFileSource extends AbstractSource implements Configurable, EventDrivenSource {

    private static final Logger LOGGER = LoggerFactory
            .getLogger(TailFileSource.class);


    /**
     * source配置参数
     */

    // 读取的配置文件
    private static String filePath;
    // 偏移量记录
    private static String offsetFile;
    // 监控的之间间隔
    private static Long interval;
    // 字符编码集
    private static String charset;

    private ExecutorService executor;

    private TailFileSourceRunner tailFileSourceRunner;



    /**
     * 读取配置文件
     *
     * @param context
     */
    @Override
    public void configure(Context context) {
        filePath = context.getString("filePath");
        offsetFile = context.getString("offsetFile");
        interval = context.getLong("interval", 1000L);
        charset = context.getString("charset");
    }

    /**
     * 启动
     */
    @Override
    public synchronized void start() {

        this.executor = Executors.newSingleThreadExecutor();
        tailFileSourceRunner = new TailFileSourceRunner(filePath, offsetFile, interval, charset, getChannelProcessor());
        executor.execute(tailFileSourceRunner);
        super.start();
    }

    /**
     * 停止
     */
    @Override
    public synchronized void stop() {

        // 停止任务线程线程
        tailFileSourceRunner.StopIt();
        executor.shutdown();

        while (!executor.isTerminated()) {
            LOGGER.debug("Waiting for exec executor service to stop");
            try {
                executor.awaitTermination(500, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                LOGGER.debug("Interrupted while waiting for exec executor service "
                        + "to stop. Just exiting.");
                Thread.currentThread().interrupt();
            }
        }


        super.stop();
    }

    /**
     * 实时监控读取文件
     */
    private static class TailFileSourceRunner implements Runnable {

        // 读取的配置文件
        private String filePath;
        // 偏移量记录
        private  File offFile;
        // 监控的之间间隔
        private Long interval;
        // charset
        private String charset;

        // Channel
        private ChannelProcessor channelProcessor;

        private RandomAccessFile raf;

        private boolean flag = Boolean.TRUE;

        /**
         * 设置是否从头开始读
         */
        private boolean startAtBeginning = false;

        private long offset = 0L;



        public TailFileSourceRunner(String filePath, String offsetFile, Long interval, String charset, ChannelProcessor channelProcessor) {
            this.filePath = filePath;
            this.interval = interval;
            this.charset = charset;
            this.channelProcessor = channelProcessor;


            offFile = new File(offsetFile);

            if (!offFile.exists()) {// 偏移量文件不存在
                try {
                    offFile.createNewFile();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            long offset = 0L;
            // 读取偏移量
            try {
                String line = FileUtils.readFileToString(offFile);
                // 偏移量
                if (line != null && !"".equals(line))   offset = Long.parseLong(line);
            } catch (IOException e) {
                LOGGER.error("读取文件偏移量失败 ",e);
            }

            //跳读文件
            try {
                raf = new RandomAccessFile(filePath, "r");
                raf.seek(offset);// 直接跳转值指定行
            } catch (FileNotFoundException e) {
                LOGGER.error("RandomAccessFile 读取文件失败 ",e);
            } catch (IOException e) {
                LOGGER.error("RandomAccessFile seek 失败 ",e);
            }

        }

        @Override
        public void run() {

           while (flag) {
               try {
                   // 读取文件
                   String logLine = new String(raf.readLine().getBytes("iso8859-1"),charset);
                   if (logLine != null) { // 读取到数据
                       final Event event = EventBuilder.withBody(logLine.getBytes(charset));
                       //发送值Chanel
                       channelProcessor.processEvent(event);
                       String rafFilePointer = raf.getFilePointer() +"";
                       // 获取文件的偏移量
                       FileUtils.writeStringToFile(offFile,rafFilePointer);
                   }else {
                       Thread.sleep(interval);
                   }
               } catch (IOException e) {
                   LOGGER.error("写入便宜量失败 ",e);
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           }

        }

        /**
         * 优雅的关闭的flume  source
         */
        public void StopIt () {
            this.flag = Boolean.FALSE;
        }

    }

}
